
package org.project.d.network.start;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollDatagramChannel;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.DatagramChannel;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

import static io.netty.channel.socket.InternetProtocolFamily.IPv4;


public abstract class NettyUDPServer extends BaseService implements Server {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    protected final int port;
    protected final String host;
    private EventLoopGroup eventLoopGroup;

    public NettyUDPServer(int port) {
        this.port = port;
        this.host = null;
    }

    public NettyUDPServer(int port, String host) {
        this.port = port;
        this.host = host;
    }

    @Override
    protected void doStart(Listener listener) throws Throwable {
        createNioServer(listener);
    }

    @Override
    protected void doStop(Listener listener) throws Throwable {
        logger.info("try shutdown {}...", this.getClass().getSimpleName());
        if (eventLoopGroup != null) eventLoopGroup.shutdownGracefully().syncUninterruptibly();
        logger.info("{} shutdown success.", this.getClass().getSimpleName());
        listener.onSuccess(port);
    }

    private void createServer(Listener listener, EventLoopGroup eventLoopGroup, ChannelFactory<? extends DatagramChannel> channelFactory) {
        this.eventLoopGroup = eventLoopGroup;
        try {
            Bootstrap b = new Bootstrap();
            b.group(eventLoopGroup)//默认是根据机器情况创建Channel,如果机器支持ipv6,则无法使用ipv4的地址加入组播
                    .channelFactory(channelFactory)
//                    .option(ChannelOption.SO_BROADCAST, true)
                    .handler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            initPipeline(pipeline);
                        }
                    });
            initOptions(b);
            ChannelFuture bind;
            if (StringUtils.isBlank(host)) {
                bind= b.bind(port);
            }else {
                bind= b.bind(new InetSocketAddress(host, port));
            }
            //直接绑定端口，指定host，收不到组播消息
            GenericFutureListener<Future<? super Void>> futureGenericFutureListener = future -> {
                if (future.isSuccess()) {
                    logger.info("udp server start success on:{}", port);
                    if (listener != null) listener.onSuccess(bind);
                } else {
                    logger.error("udp server start failure on:{}", port, future.cause());
                    if (listener != null) listener.onFailure(future.cause());
                }
            };
            bind.addListener(futureGenericFutureListener);


        } catch (Exception e) {
            logger.error("udp server start exception", e);
            if (listener != null) listener.onFailure(e);
            throw new ServiceException("udp server start exception, port=" + port, e);
        }
    }

    private void createNioServer(Listener listener) {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(
                1, new DefaultThreadFactory("project-udp-server")
        );
        eventLoopGroup.setIoRatio(100);
        createServer(listener, eventLoopGroup, () -> new NioDatagramChannel(IPv4));//默认是根据机器情况创建Channel,如果机器支持ipv6,则无法使用ipv4的地址加入组播
    }

    @SuppressWarnings("unused")
    private void createEpollServer(Listener listener) {
        EpollEventLoopGroup eventLoopGroup = new EpollEventLoopGroup(
                1, new DefaultThreadFactory("project-udp-server")
        );
        eventLoopGroup.setIoRatio(100);
        createServer(listener, eventLoopGroup, EpollDatagramChannel::new);
    }

    protected void initOptions(Bootstrap b) {
        b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        b.option(ChannelOption.SO_REUSEADDR, true);
        b.option(ChannelOption.SO_BROADCAST, true);
    }

    protected abstract void initPipeline(ChannelPipeline channelPipeline);

}
